package com.etl.dataflow.scheduler.communication.service;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import com.etl.dataflow.common.entity.ExecutorInfo;
import com.etl.dataflow.common.entity.TaskInfo;
import com.etl.dataflow.common.enums.TaskStateEnum;
import com.etl.dataflow.common.exception.ServiceException;
import com.etl.dataflow.common.response.Code;
import com.etl.dataflow.common.response.R;
import com.etl.dataflow.scheduler.admin.entity.DataflowExecutorInfo;
import com.etl.dataflow.scheduler.admin.entity.DataflowTaskRecord;
import com.etl.dataflow.scheduler.admin.mapper.DataflowExecutorInfoMapper;
import com.etl.dataflow.scheduler.admin.mapper.DataflowTaskRecordMapper;
import com.etl.dataflow.scheduler.communication.ExecutorStateChecker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * 执行器注册接口service
 *
 * @author dx
 * @since 2022/4/20
 */
@Service
@Slf4j
public class SchedulerService {

    private static final Set<Integer> UN_UPDATABLE_STATES = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
            TaskStateEnum.TERMINATED.getValue(),
            TaskStateEnum.FINISHED.getValue(),
            TaskStateEnum.FATAL_ERROR.getValue(),
            TaskStateEnum.TIMEOUT.getValue()
    )));

    private final DataflowTaskRecordMapper taskRecordMapper;

    private final DataflowExecutorInfoMapper executorInfoMapper;

    private final ExecutorStateChecker executorStateChecker;

    public SchedulerService(DataflowTaskRecordMapper taskRecordMapper, DataflowExecutorInfoMapper executorInfoMapper, ExecutorStateChecker executorStateChecker) {
        this.taskRecordMapper = taskRecordMapper;
        this.executorInfoMapper = executorInfoMapper;
        this.executorStateChecker = executorStateChecker;
    }

    /**
     * 执行器注册
     *
     * @param executorInfo 执行器信息
     * @return R
     */
    public R register(ExecutorInfo executorInfo) {
        DataflowExecutorInfo dataflowExecutorInfo = new DataflowExecutorInfo();
        BeanUtil.copyProperties(executorInfo, dataflowExecutorInfo);
        int effectRows = executorInfoMapper.updateById(dataflowExecutorInfo);
        if (effectRows == 0) {
            effectRows = executorInfoMapper.insert(dataflowExecutorInfo);
        }
        if (effectRows >= 0) {
            executorStateChecker.register(executorInfo);
            return R.success()
                    .put("timestamp", System.currentTimeMillis())
                    .put("executor", executorInfo);
        }
        throw new ServiceException(Code.TASK_NOT_FOUND_CANNOT_TERMINATE);
    }

    public R updateTaskState(String uuid, String taskId, String taskRecordId, TaskStateEnum newState) {
        log.debug("更新任务状态，taskRecordId: {}, newState: {}", taskRecordId, newState.getMsg());
        // 之所以直接更新数据库而不通过ExecutorStateChecker更新执行器任务，是因为它会自动请求执行器的运行中任务接口刷新任务列表
        DataflowTaskRecord taskRecord = taskRecordMapper.selectById(taskRecordId);
        if (taskRecord == null) {
            log.debug("更新任务状态失败，任务记录不存在，taskRecordId: {}, newStateValue: {}, newState: {}", taskRecordId, newState.getValue(), newState.getMsg());
            throw new ServiceException(Code.TASK_NOT_FOUND_CANNOT_UPDATE);
        }
        if (UN_UPDATABLE_STATES.contains(taskRecord.getTaskState())) {
            log.debug("更新任务状态失败，任务已经处于'{}'状态，taskRecordId: {}, newStateValue: {}, newState: {}",
                    TaskStateEnum.of(taskRecord.getTaskState()).getMsg(), taskRecordId, newState.getValue(), newState.getMsg());
            throw new ServiceException(Code.TASK_STATE_UNMODIFIABLE);
        }
        Long recordId = taskRecord.getId();
        taskRecord = new DataflowTaskRecord();
        taskRecord.setId(recordId);
        taskRecord.setTaskState(newState.getValue());
        if (newState.equals(TaskStateEnum.FATAL_ERROR)) {
            taskRecord.setErrorLog(newState.getMsg());
        }
        if (newState.equals(TaskStateEnum.FINISHED) ||
                newState.equals(TaskStateEnum.TERMINATED) ||
                newState.equals(TaskStateEnum.FATAL_ERROR)) {
            taskRecord.setEndTime(new Date());
        }
        taskRecordMapper.updateById(taskRecord);
        taskRecord = taskRecordMapper.selectById(taskRecordId);
        boolean success = taskRecord.getTaskState() == newState.getValue();
        if (success) {
            log.debug("更新任务状态成功，taskRecordId: {}, newStateValue: {}, newState: {}", taskRecordId, newState.getValue(), newState.getMsg());
        } else {
            log.debug("更新任务状态失败，taskRecordId: {}, newStateValue: {}, newState: {}", taskRecordId, newState.getValue(), newState.getMsg());
        }
        updateTaskStateInMemory(uuid, taskId, newState);
        return R.success().put("result", success);
    }

    private void updateTaskStateInMemory(String uuid, String taskId, TaskStateEnum newState) {
        ExecutorInfo executorInfo = executorStateChecker.getExecutorInfoByUuid(uuid);
        if (executorInfo == null) {
            return;
        }
        Iterator<TaskInfo> iterator = executorInfo.getTaskList().iterator();
        TaskInfo needRemoveTask = null;
        while (iterator.hasNext()) {
            TaskInfo taskInfo = iterator.next();
            if (!StrUtil.equals(taskInfo.getId(), taskId)) {
                continue;
            }
            taskInfo.setState(newState);
            if (TaskStateEnum.TERMINATED.equals(newState)
                    || TaskStateEnum.FINISHED.equals(newState)
                    || TaskStateEnum.FATAL_ERROR.equals(newState)) {
                needRemoveTask = taskInfo;
            }
            break;
        }
        executorInfo.getTaskList().remove(needRemoveTask);
    }

    public R updateTaskStateToFinished(String uuid, String taskId, String taskRecordId) {
        return this.updateTaskState(uuid, taskId, taskRecordId, TaskStateEnum.FINISHED);
    }

    public R updateTaskStateToError(String uuid, String taskId, String taskRecordId, String errorMsg) {
        TaskStateEnum fatalError = TaskStateEnum.FATAL_ERROR;
        fatalError.setMsg(errorMsg);
        return this.updateTaskState(uuid, taskId, taskRecordId, fatalError);
    }

    public R updateTaskStateToTimeout(String uuid, String taskId, String taskRecordId) {
        return this.updateTaskState(uuid, taskId, taskRecordId, TaskStateEnum.TIMEOUT);
    }
}
